package com.denghq.projectbuilder.component.consul.config.configuration;

import com.denghq.projectbuilder.component.config.metadata.ConfigDescriptor;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.QueryParams;
import com.ecwid.consul.v1.Response;
import com.ecwid.consul.v1.kv.model.GetValue;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.micrometer.core.annotation.Timed;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.endpoint.event.RefreshEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.style.ToStringCreator;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 配置变更检测器
 */
@Slf4j
public class ConsulConfigWatcher implements ApplicationEventPublisherAware, SmartLifecycle {

    private final TaskScheduler taskScheduler;
    private ScheduledFuture<?> watchFuture;
    private final ConsulConfigProperties properties;
    private final ConsulClient consul;
    private LinkedHashMap<String, Long> consulIndexes;
    private final AtomicBoolean running;
    private boolean firstTime;
    private ApplicationEventPublisher publisher;

    private final ExecutorService cachedThreadPool;
    //保存当前正在加载的context,如果正在加载则跳过提交任务，主要解决获取context没有变化的时候会等待55s的问题。
    private final List<String> currentLoadingContextList = Lists.newLinkedList();


    public ConsulConfigWatcher(ConsulConfigProperties properties, ConsulClient consul, List<ConfigDescriptor> configDescriptors) {
        this(properties, consul, configDescriptors, getTaskScheduler());
    }

    private static ThreadPoolTaskScheduler getTaskScheduler() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.initialize();
        return taskScheduler;
    }

    public ConsulConfigWatcher(ConsulConfigProperties properties, ConsulClient consul, List<ConfigDescriptor> configDescriptors, TaskScheduler taskScheduler) {
        this.running = new AtomicBoolean(false);
        this.firstTime = true;
        this.properties = properties;
        this.consul = consul;
        this.consulIndexes = getConsulIndexes(configDescriptors);
        this.taskScheduler = taskScheduler;
        this.cachedThreadPool = Executors.newFixedThreadPool(this.consulIndexes.size());
    }

    private LinkedHashMap<String, Long> getConsulIndexes(List<ConfigDescriptor> configDescriptors) {
        LinkedHashMap<String, Long> res = Maps.newLinkedHashMap();
        if (!CollectionUtils.isEmpty(configDescriptors)) {
            configDescriptors.forEach(o ->
                    res.put(o.getCategoryNo(), -1L)
            );
        }
        return res;
    }

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.publisher = applicationEventPublisher;
    }


    @Timed("consul.watch-config-keys")
    public void watchConfigKeyValues() {
        if (this.running.get()) {
            Iterator var1 = this.consulIndexes.keySet().iterator();

            while (var1.hasNext()) {
                String context = (String) var1.next();
                /*if (!context.endsWith("/")) {
                    context = context + "/";
                }*/
                synchronized (currentLoadingContextList) {
                    if (!currentLoadingContextList.contains(context)) {
                        currentLoadingContextList.add(context);
                        cachedThreadPool.submit(() -> loadContext(context));
                    }
                }
            }
        }

        this.firstTime = false;
    }

    private void loadContext(String context) {
        try {
            Long currentIndex = (Long) this.consulIndexes.get(context);
            if (currentIndex == null) {
                currentIndex = -1L;
            }

            log.trace("watching consul for context '" + context + "' with index " + currentIndex);
            String aclToken = this.properties.getAclToken();
            if (StringUtils.isEmpty(aclToken)) {
                aclToken = null;
            }

            Response<List<GetValue>> response = this.consul.getKVValues(context, aclToken, new QueryParams((long) this.properties.getWatch().getWaitTime(), currentIndex));
            if (response.getValue() != null && !((List) response.getValue()).isEmpty()) {
                Long newIndex = response.getConsulIndex();
                if (newIndex != null && !newIndex.equals(currentIndex)) {
                    if (!this.consulIndexes.containsValue(newIndex) && !currentIndex.equals(-1L)) {
                        log.trace("Context " + context + " has new index " + newIndex);
                        log.warn("Context " + context + " has new index " + newIndex);
                        RefreshEventData data = new RefreshEventData(context, currentIndex, newIndex);
                        this.publisher.publishEvent(new RefreshEvent(this, data, data.toString()));
                    } else if (log.isTraceEnabled()) {
                        log.warn("Event for index already published for context " + context);
                        log.trace("Event for index already published for context " + context);
                    }

                    this.consulIndexes.put(context, newIndex);
                } else if (log.isTraceEnabled()) {
                    log.trace("Same index for context " + context);
                }
            } else if (log.isTraceEnabled()) {
                log.trace("No value for context " + context);
            }
        } catch (Exception var8) {
            if (this.firstTime && this.properties.isFailFast()) {
                log.error("Fail fast is set and there was an error reading autoconfigure from consul.");
                ReflectionUtils.rethrowRuntimeException(var8);
            } else if (log.isTraceEnabled()) {
                log.trace("Error querying consul Key/Values for context '" + context + "'", var8);
            } else if (log.isWarnEnabled()) {
                log.warn("Error querying consul Key/Values for context '" + context + "'. Message: " + var8.getMessage());
            }
        }
        synchronized (currentLoadingContextList) {
            currentLoadingContextList.remove(context);
        }
    }

    static class RefreshEventData {
        private final String context;
        private final Long prevIndex;
        private final Long newIndex;

        public RefreshEventData(String context, Long prevIndex, Long newIndex) {
            this.context = context;
            this.prevIndex = prevIndex;
            this.newIndex = newIndex;
        }

        public String getContext() {
            return this.context;
        }

        public Long getPrevIndex() {
            return this.prevIndex;
        }

        public Long getNewIndex() {
            return this.newIndex;
        }

        public boolean equals(Object o) {
            if (this == o) {
                return true;
            } else if (o != null && this.getClass() == o.getClass()) {
                ConsulConfigWatcher.RefreshEventData that = (ConsulConfigWatcher.RefreshEventData) o;
                return Objects.equals(this.context, that.context) && Objects.equals(this.prevIndex, that.prevIndex) && Objects.equals(this.newIndex, that.newIndex);
            } else {
                return false;
            }
        }

        public int hashCode() {
            return Objects.hash(new Object[]{this.context, this.prevIndex, this.newIndex});
        }

        public String toString() {
            return (new ToStringCreator(this)).append("context", this.context).append("prevIndex", this.prevIndex).append("newIndex", this.newIndex).toString();
        }
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void stop(Runnable callback) {
        this.stop();
        callback.run();
    }

    @Override
    public void start() {
        if (this.running.compareAndSet(false, true)) {
            this.watchFuture = this.taskScheduler.scheduleWithFixedDelay(this::watchConfigKeyValues, (long) this.properties.getWatch().getDelay());
        }
    }

    @Override
    public void stop() {
        if (this.running.compareAndSet(true, false) && this.watchFuture != null) {
            this.watchFuture.cancel(true);
        }
    }

    @Override
    public boolean isRunning() {
        return this.running.get();
    }

    @Override
    public int getPhase() {
        return 0;
    }
}
